#!/usr/bin/env python
# -*- coding: utf-8 -*-
__title__ = '订阅ctp行情并入库'
__author__ = 'HaiFeng'
__mtime__ = '20180723'

import threading
import sys
import time
import json
import getpass
from time import sleep
from datetime import datetime, timedelta
from py_ctp.enums import InstrumentStatus
from py_ctp.trade import CtpTrade
from py_ctp.quote import CtpQuote
from py_ctp.structs import InfoField, Tick, InstrumentField
from queue import Queue
import config as cfg
from itertools import groupby


class TickCtp(object):
    """"""

    def __init__(self):
        """初始化"""
        self.inst_mins = {}
        self.received_tick = []
        self.trade_time = {}
        self.trading_days = []

        # 合约
        self.instruments = {}

        self.TradingDay = ''
        self.Actionday = ''
        self.Actionday1 = ''
        self.tick_time = ''

        # 计算指数相关
        self.product_time = {}
        # self.product_rate = {}

        self.inst_queue = Queue(0)

        self.q = CtpQuote()

    def start_quote(self):
        """"""
        # 隔夜时:行情重新登录后会重新获取夜盘的数据
        # if len(self.inst_pre_vol) == 0:
        cfg.log.info('start quote')
        self.q.OnConnected = self.q_OnFrontConnected
        self.q.OnDisconnected = self.q_OnDisConnected
        self.q.OnUserLogin = self.q_OnRspUserLogin
        self.q.OnTick = self.q_OnTick

        self.q.ReqConnect(cfg.front_quote)

    def q_OnFrontConnected(self, obj):
        # cfg.log.war('q:connected')
        self.q.ReqUserLogin(cfg.investor, cfg.pwd, cfg.broker)

    def q_OnDisConnected(self, obj, nReason: int):
        if nReason != 4097:
            cfg.log.war('q:disconnected')

    def _get_instruments(self):
        """取合约信息"""
        self.instruments.clear()
        df: DataFrame = cfg.mysql_df(cfg.sql_instruments)

        for i, r in df.iterrows():
            info = InstrumentField()
            info.InstrumentID = r['InstrumentID']
            info.ExchangeID = r['ExchangeID']
            info.MaxOrderVolume = r['MaxOrderVolume']
            info.PriceTick = r['PriceTick']
            info.ProductID = r['ProductID']
            info.ProductType = r['ProductType']
            info.VolumeMultiple = r['VolumeMultiple']
            self.instruments[info.InstrumentID] = info

    def _get_trading_time(self):
        """取各合约交易时间"""
        self.trade_time.clear()
        times = []
        g = cfg.mysql_fetchall(cfg.sql_time)
        times = [{t[0]: json.loads(t[1])} for t in g]

        # 按时间排序, 确保最后实施的时间段作为依据.
        # 根据时间段设置,生成 opens; ends; mins盘中时间
        for group in times:
            g_id = list(group.keys())[0]
            section = list(group.values())[0]
            opens = []
            ends = []
            mins = []
            for s in section:
                opens.append((datetime.strptime(s['Begin'], '%H:%M:%S') + timedelta(minutes=-1)).strftime('%H:%M:00'))
                ends.append(s['End'])
                t_begin = datetime.strptime('20180101' + s['Begin'], '%Y%m%d%H:%M:%S')
                s_end = datetime.strptime('20180101' + s['End'], '%Y%m%d%H:%M:%S')
                if t_begin > s_end:  # 夜盘
                    s_end += timedelta(days=1)
                while t_begin < s_end:
                    mins.append(t_begin.strftime('%H:%M:00'))
                    t_begin = t_begin + timedelta(minutes=1)
            self.trade_time[g_id] = {'Opens': opens, 'Ends': ends, 'Mins': mins}

    def _init_quote(self):
        """初始化行情参数"""
        # 取合约信息
        self._get_instruments()
        # 获取品种交易时间
        self._get_trading_time()
        # 取交易日
        days = cfg.mysql_fetchall(cfg.sql_trade_date)
        self.trading_days = [r[0] for r in days]

        # 当前交易日
        cur_day = time.strftime('%Y%m%d', time.localtime())
        self.TradingDay = cur_day
        days = [x for x in self.trading_days if x >= self.TradingDay]
        cur_hour = time.localtime().tm_hour
        if cur_hour >= 16:
            self.TradingDay = days[1]
            if cur_hour <= 20:
                # 清理数据
                cfg.rds.flushdb()
        # 行情日期
        self.Actionday = self.TradingDay if self.trading_days.index(self.TradingDay) == 0 else self.trading_days[self.trading_days.index(self.TradingDay) - 1]
        self.Actionday1 = (datetime.strptime(self.Actionday, '%Y%m%d') + timedelta(days=1)).strftime('%Y%m%d')

        # 品种交易时间==>合约交易时间
        for inst, info in self.instruments.items():
            if info.ProductID in self.trade_time:
                self.trade_time[inst] = self.trade_time[info.ProductID]
            else:
                self.trade_time[inst] = self.trade_time['default']

    def q_OnRspUserLogin(self, obj, info: InfoField):
        cfg.log.info('q:{}'.format(info))

        # 数据同步到ali-mysql
        # cfg.oracle_2_mysql('t_instrument')

        self._init_quote()

        lst = sorted(self.instruments.values(), key=lambda i: i.ProductID)
        for p, v in groupby(lst, key=lambda i: i.ProductID):
            # for p, v in self.trade_time.items():
            self.inst_mins[f'{p}_000'] = {'pre_vol': 0}

        for inst in self.instruments.keys():
            self.inst_mins[inst] = {'pre_vol': 0}
            self.q.ReqSubscribeMarketData(inst)

        # self.inst_mins['rb1910'] = {}
        # self.q.ReqSubscribeMarketData('rb1910')
        cfg.log.info('sub count:{}'.format(len(self.inst_mins)))

    def q_OnTick(self, obj: CtpQuote, tick: Tick):
        # 某个合约报 most recent call last 错误: 尝试改为线程处理
        if sys.float_info.max == tick.LastPrice:  # or sys.float_info.max == tick.AskPrice or sys.float_info.max == tick.BidPrice or sys.float_info.max == tick.LowerLimitPrice:
            return
        threading.Thread(target=self.run_tick, args=(tick,)).start()
        # 非线程模式,导致数据延时
        # self.run_tick(tick)

    def run_tick(self, tick: Tick):
        # 对tick时间进行修正处理
        ut = tick.UpdateTime[0:6] + '00'
        mins_dict = self.trade_time[tick.Instrument]
        # 由下面的 updatetime[-2:0] != '00' 处理
        if ut not in mins_dict['Mins']:
            # 开盘/收盘
            if ut in mins_dict['Opens']:
                ut = (datetime.strptime(ut, '%H:%M:%S') + timedelta(minutes=1)).strftime('%H:%M:%S')
            elif ut in mins_dict['Ends']:
                # 重新登录会收到上一节的最后tick
                tick_dt = datetime.strptime('{} {}'.format(datetime.now().strftime('%Y%m%d'), tick.UpdateTime), '%Y%m%d %H:%M:%S')
                now_dt = datetime.now()
                diff_snd = 0
                if tick_dt > now_dt:
                    diff_snd = (tick_dt - now_dt).seconds
                else:
                    diff_snd = (now_dt - tick_dt).seconds
                if diff_snd > 30:
                    return
                ut = (datetime.strptime(ut, '%H:%M:%S') + timedelta(minutes=-1)).strftime('%H:%M:%S')
            else:
                return
        # 首tick不处理(新开盘时会收到之前的旧数据)
        if tick.Instrument not in self.received_tick:
            self.received_tick.append(tick.Instrument)
            return

        inst_info: InstrumentField = self.instruments.get(tick.Instrument)
        if not inst_info:
            return
        # 合约合成分钟
        self.tick_min(tick, ut)

        # 指数合约合成分成
        pre_time = self.product_time.get(inst_info.ProductID)
        if not pre_time:
            self.product_time[inst_info.ProductID] = tick.UpdateTime
        elif pre_time != tick.UpdateTime and tick.UpdateTime[-2:0] != '00':  # 整分时等下一秒再处理,以处理小节收盘单tick的问题
            self.product_time[inst_info.ProductID] = tick.UpdateTime

            # 计算合约权重
            ticks = [f for k, f in self.q.inst_tick.items() if self.instruments[k].ProductID == inst_info.ProductID]
            sum_oi = sum([f.OpenInterest for f in ticks])
            if sum_oi == 0:
                return

            rate = json.loads('{{{}}}'.format(','.join(['"{}":{}'.format(f.Instrument, f.OpenInterest / sum_oi) for f in ticks])))
            # 计算000
            tick000: Tick = Tick()
            tick000.Instrument = inst_info.ProductID + '_000'
            tick000.UpdateTime = tick.UpdateTime
            for inst, rate in rate.items():
                f: Tick = self.q.inst_tick[inst]
                tick000.AskPrice += f.AskPrice * rate
                tick000.BidPrice += f.BidPrice * rate
                tick000.LastPrice += f.LastPrice * rate
                tick000.AveragePrice += f.AveragePrice * rate
                tick000.LowerLimitPrice += f.LowerLimitPrice * rate
                tick000.UpperLimitPrice += f.UpperLimitPrice * rate
                tick000.AskVolume += f.AskVolume
                tick000.BidVolume += f.BidVolume
                tick000.Volume += f.Volume
                tick000.OpenInterest += f.OpenInterest
            # 防止因值为 sys.float_info.max 而报错: 只有lastprice参与分钟数据计算
            # tick000.AskPrice = round(tick000.AskPrice / inst.PriceTick) * inst.PriceTick
            # tick000.BidPrice = round(tick000.BidPrice / inst.PriceTick) * inst.PriceTick
            tick000.LastPrice = round(tick000.LastPrice / inst_info.PriceTick) * inst_info.PriceTick
            # tick000.AveragePrice = round(tick000.AveragePrice / inst.PriceTick) * inst.PriceTick
            # tick000.LowerLimitPrice = round(tick000.LowerLimitPrice / inst.PriceTick) * inst.PriceTick
            # tick000.UpperLimitPrice = round(tick000.UpperLimitPrice / inst.PriceTick) * inst.PriceTick
            self.tick_min(tick000, ut)

    def tick_min(self, tick: Tick, ut: str):
        actionday = self.TradingDay
        if ut[0:2] > '20':
            actionday = self.Actionday
        elif ut[0:2] < '04':
            actionday = self.Actionday1
        # 分钟入库
        cur_min = self.inst_mins[tick.Instrument]

        if '_id' not in cur_min or cur_min['_id'][11:] != ut:
            cur_min['_id'] = actionday[0:4] + '-' + actionday[4:6] + '-' + actionday[6:] + ' ' + ut
            cur_min['TradingDay'] = self.TradingDay
            cur_min['Low'] = cur_min['Close'] = cur_min['High'] = cur_min['Open'] = tick.LastPrice
            cur_min['OpenInterest'] = tick.OpenInterest
            # cur_min['AveragePrice'] = tick.AveragePrice
            # 首个tick不计算成交量, 否则会导致隔夜的早盘第一个分钟的成交量非常大
            cur_min['Volume'] = tick.Volume - cur_min['pre_vol']
            cur_min['pre_vol'] = tick.Volume
            if cfg.rds:
                cfg.rds.rpush(tick.Instrument, json.dumps(cur_min))
        else:
            cur_min['High'] = max(cur_min['High'], tick.LastPrice)
            cur_min['Low'] = min(cur_min['Low'], tick.LastPrice)
            cur_min['Close'] = tick.LastPrice
            cur_min['OpenInterest'] = tick.OpenInterest
            cur_min['Volume'] = tick.Volume - cur_min['pre_vol']
            if cfg.rds:
                cfg.rds.lset(tick.Instrument, -1, json.dumps(cur_min))
        self.tick_time = tick.UpdateTime


if __name__ == '__main__':
    TickCtp().start_quote()
    while True:
        sleep(60)
